﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MassTransit;

namespace MassTransitDemo.JobService.Consumers
{
    public class VideoConvertedConsumer : IConsumer<VideoConverted>
    {
        static readonly ConcurrentDictionary<string, GroupInfo> _groupInfo = new();

        public Task Consume(ConsumeContext<VideoConverted> context)
        {
            var groupInfo = _groupInfo.GetOrAdd(context.Message.GroupId, _ => new GroupInfo(context.Message.Count));

            var received = groupInfo.Increment(context.Message.Index);
            if (received == groupInfo.Count)
            {
                Console.WriteLine("Group Completed: {GroupId} ({Count})", context.Message.GroupId, groupInfo.Count);
            }
            else if (groupInfo.Count - received < 10)
            {
                Console.WriteLine("Group Remaining: {GroupId} ({Count}) {Remaining}", context.Message.GroupId, groupInfo.Count,
                    string.Join(", ", Enumerable.Range(0, groupInfo.Count).Except(groupInfo.Indices)));
            }
            else
            {
                Console.WriteLine("Group Progress: {GroupId} ({Received}/{Count})", context.Message.GroupId, received, groupInfo.Count);
            }

            return Task.CompletedTask;
        }

        class GroupInfo
        {
            public readonly int Count;
            public readonly HashSet<int> Indices;
            public int Received;

            public GroupInfo(int count)
            {
                Count = count;
                Indices = new HashSet<int>();
            }

            public int Increment(int index)
            {
                Indices.Add(index);

                return Interlocked.Increment(ref Received);
            }
        }
    }

    public interface VideoConverted
    {
        string GroupId { get; }
        int Index { get; }
        int Count { get; }
    }
}
